package FocusCrawler;
require Exporter;
our @ISA       = qw(Exporter);
our @EXPORT_OK = qw(
    &get_canonical_url &get_new_url &get_host
    &is_target_anchor 
    &do_leaf_numbering_recursive &collect_features
    %params $frontier_db $frontier_cnt $target_cnt $failure_cnt
    %urls %dumps %digests &check_db
);
our $VERSION = 1.00;

use warnings;
use strict;
use POE;
use POE::Component::Client::DNS;
use POE::Component::Client::HTTP;
use BerkeleyDB;
use MLDBM qw(DB_File Storable);
use YAML;
use DB_File;
use Storable;
use File::Spec;
use Encode;
use Digest::MD5;
use URI;
use HTTP::Request;
use HTML::Parser;
use HTML::TreeBuilder;
use HTML::Tagset;
use WWW::RobotRules::AnyDBM_File;
use TFIDF;

use Readonly;
Readonly our $SEED          => 2;
Readonly our $ROBOT         => 3;
Readonly our $REDIRECT      => 4;
Readonly our $POCO_HTTP     => 'ua';
Readonly our $URL_FETCHED   => 'done';
Readonly our $RULE_SUCCESS  => 'success';
Readonly our $RULE_FAILURE  => 'failure';

our %params;
our ( $frontier_db, $frontier_cnt );
our ( %urls, %dumps, %digests );
our ( $target_cnt, $failure_cnt);
my $epoch;
my ( %suffixes, %classifier );
my $svm_bias;
my $config_file;
my $download_cnt;
my @stat;
my %busy_hosts;
my $busy_hosts_cnt;
my $event_loop_flag;
my %frontier;
my $log_fh;
my $rules;

sub init {
    (undef, my $params_ref) = @_;
    %params = %$params_ref;
    (my $base_dir = $params{base_dir} ) =~ s/^~/$ENV{HOME}/;
    $base_dir = File::Spec->rel2abs($params{base_dir});
    chdir $base_dir or die "$!";
    $params{$_} = File::Spec->rel2abs($params{$_}) for qw(classifier_file seeds_file);
    my $working_dir = "$base_dir/$params{working_dir}";
    mkdir $working_dir unless -d $working_dir;
    chdir $working_dir or die "$!";
    $config_file = 'config';
    YAML::DumpFile($config_file, $params_ref);
    $epoch       = ( stat $config_file )[9];
    %suffixes   = %{ YAML::LoadFile( $params{suffixes_yml_file} ) };
    %classifier = %{ Storable::retrieve( $params{classifier_file} ) };
    $svm_bias = $classifier{BIAS};
}

sub run {
    my $package = shift;
    POE::Component::Client::HTTP->spawn(
        Protocol => 'HTTP/1.1',
        Alias    => $POCO_HTTP,
        Timeout  => $params{time_out},
        Agent    => $params{user_agent},
        MaxSize  => $params{max_size},
    );

    POE::Session->create(
        package_states => [
            $package =>
                [qw(_start _stop ready got_sigint got_sigtstp got_response)],
        ],
    );

    POE::Kernel->run();
    exit(0);
}

sub _start {
    my $package = $_[OBJECT];

    $download_cnt = $target_cnt = $failure_cnt = 0;
    $busy_hosts_cnt  = 0;
    $event_loop_flag = 1;

    if ($params{learn_flag}) {
        $package->learn() if $package->can('learn');
    }
    else {
        $package->restore_apprentice() if $package->can('restore_apprentice');
    }

    unless ($params{resume_flag}) {
        unlink $params{$_} for qw(url_db_file dump_db_file digest_db_file frontier_db_file log_file statistics_file);
        unlink "$params{rule_db_file}.$_" for qw(dir pag);
    }

    $rules =
        new WWW::RobotRules::AnyDBM_File @params{ 'user_agent',
        'rule_db_file' };

    tie %urls,    'MLDBM',   $params{url_db_file}    or die "$!";
    tie %dumps,   'MLDBM',   $params{dump_db_file}   or die "$!";
    tie %digests, 'DB_File', $params{digest_db_file} or die "$!";
    $frontier_db = tie %frontier, 'BerkeleyDB::Btree',
        -Filename => $params{frontier_db_file},
        -Flags    => DB_CREATE,
        -Property => DB_DUP,
        -Compare  => sub { $_[1] <=> $_[0] },
        or die "$!";

    $frontier_cnt = scalar keys %frontier;

    if ($params{resume_flag}) {
        $package->reorder_frontier() if $params{learn_flag};
        @stat = @{ Storable::retrieve( $params{statistics_file} ) };
        $download_cnt = scalar @stat;
        for (@stat) {
            $target_cnt++  if $_ eq '+';
            $failure_cnt++ if $_ eq '-';
        }
        open $log_fh, ">> $params{log_file}" or die "$!";
        print STDERR "\nPrevious state restored: $frontier_cnt\n\n" unless $params{silent};
    }
    else {
        open $log_fh, "> $params{log_file}" or die "$!";
        $package->init_frontier();
    }

    my $old_fh = select $log_fh;
    $| = 1;
    select $old_fh;

    $_[KERNEL]->sig( INT  => 'got_sigint' );
    $_[KERNEL]->sig( TSTP => 'got_sigtstp' );
    $_[KERNEL]->yield('ready');
}

sub _stop {
    my $package = $_[OBJECT];
    $package->display_state(*STDERR) unless $params{silent};
    $package->display_state($log_fh);
    untie %urls;
    untie %dumps;
    untie %digests;
    undef $frontier_db;
    untie %frontier;
    close $log_fh;
    Storable::store( \@stat, $params{statistics_file} );
    $package->destroy() if $package->can('destroy');
}

sub got_sigint {
    print STDERR "\n**** Got signal of quit, please wait... ****\n\n" unless $params{silent};
    $event_loop_flag = 0;
    $_[KERNEL]->sig_handled();
}

sub got_sigtstp {
    my $package = $_[OBJECT];
    $package->display_state(*STDERR);
    my $new_epoch = ( stat $config_file )[9];
    if ( $new_epoch > $epoch ) {
        $epoch  = $new_epoch;
        %params = %{ YAML::LoadFile($config_file) };
        unless ($params{silent}) {
            print STDERR "\n**** New Parameters folded in! ****\n\n";
            print STDERR YAML::Dump( \%params );
            print STDERR '*' x 50, "\n";
        }
    }
    $_[KERNEL]->sig_handled();
}

sub ready {
    my $package = $_[OBJECT];
    return unless $event_loop_flag;
    return if defined $params{max_download} and $target_cnt + $failure_cnt > $params{max_download} or defined $params{max_pos_download} and $target_cnt  > $params{max_pos_download};
    return unless $busy_hosts_cnt > 0 or $frontier_cnt > 0;
    if ( $busy_hosts_cnt < $params{max_http_comp} and $frontier_cnt > 0 ) {
        my ($priority, $url)  = 0;
        my $cursor = $frontier_db->db_cursor();
        while ( $cursor->c_get( $priority, $url, DB_NEXT ) == 0
            and $busy_hosts_cnt < $params{max_http_comp} )
        {
            #just for FirstOrder module's convienience
            if (not ref $urls{$url}) {
                $cursor->c_del();
                next;
            }
            if ($package->can('prefetch_filter')) {
                last if $package->prefetch_filter($priority);
            }
            my $host = &get_host($url);
            next if exists $busy_hosts{$host};
            my $info = $urls{$url};
            my $robot_file = "http://$host/robots.txt";
            my ( $job, $del_flag );
            my $status = $urls{$robot_file};
            if ( not defined $status ) {
                $job = [ $robot_file, time, undef, $ROBOT ];
                $del_flag = 0;
            }
            elsif ($status eq $RULE_FAILURE
                or $status eq $RULE_SUCCESS and $rules->allowed($url) )
            {
                $job = [ $url, time, $info, $priority ];
                $del_flag = 1;
            }
            else {
                print $log_fh "$url was disallowed by robots.txt\n";
                $job      = undef;
                $del_flag = 1;
            }
            if ( defined $job ) {
                my $new_url = $job->[0];
                my $header  = HTTP::Headers->new;
                $header->header(
                    'Content-Type'     => 'text/html',
                    'Content-Language' => 'en-US'
                ) unless $new_url eq $robot_file;
                my $request = HTTP::Request->new( GET => $new_url, $header );
                print $log_fh "\nrequest: $new_url\n";
                $_[KERNEL]
                    ->post( $POCO_HTTP, 'request', 'got_response', $request,
                    $job );
                $busy_hosts{$host} = $job;
                $busy_hosts_cnt++;
            }
            if ($del_flag) {
                $package->post_check($url, $info) if $package->can('post_check');
                $urls{$url} = $URL_FETCHED;
                $cursor->c_del();
                $frontier_cnt--;
            }
        }
        $cursor->c_close();
    }
}

sub got_response {
    my $now = time;
    my ( $package, $job, $response ) =
        ( $_[OBJECT], $_[ARG0][1], $_[ARG1][0] );
    my ( $url, $request_time, $info, $priority ) = @$job;
    my $host         = &get_host($url);
    my $elapsed_time = $now - $request_time;

    my ( $content, $base ) = ( $response->content, $response->base );

    if ( not defined $info ) {
        if ( $response->is_success and $content ) {
            $rules->parse( $url, $content );
            $urls{$url} = $RULE_SUCCESS;
        }
        else {
            $urls{$url} = $RULE_FAILURE;
        }
    }
    else {
        my ( $history, $path, @others )      = @$info;
        my ( $from,    @redirects ) = @$history;
        print $log_fh "\nresponse: $url [$elapsed_time]\n";
        print $log_fh "status: ", $response->status_line, "\n";
        print $log_fh "path: $path\n";
        print $log_fh "priority: $priority\n";
        print $log_fh "redirects: ", join( " => ", @redirects ), "\n"
            if @redirects;

        if ( $response->is_success ) {
            {
                my $content_type = $response->header('Content-Type');
                if ( $content_type and $content_type !~ m{^text/html|^\s*$}i )
                {
                    print $log_fh
                        "warn: not text/html content-type: $content_type\n";
                    last;
                }
                my $content_lang = $response->header('Content-Language');
                if ( $content_lang and $content_lang !~ /^en|^\s*$/i ) {
                    print $log_fh
                        "warn: not english content-language: $content_lang\n";
                    last;
                }

                unless ($content) {
                    print $log_fh "warn: empty content\n";
                    last;
                }

                my $no_follow_flag = 0;
                my $html_parser    = HTML::Parser->new(
                    api_version     => 3,
                    marked_sections => 1
                );
                $html_parser->ignore_elements(qw(script style));
                $html_parser->handler( text => [], '@{dtext}' );
                $html_parser->report_tags(qw(html meta frameset));
                $html_parser->handler(
                    start => sub {
                        my ( $self, $tag, $attr ) = @_;
                        if (    $tag eq 'html'
                            and $attr->{lang}
                            and $attr->{lang} !~ /^en/i )
                        {
                            print $log_fh
                                sprintf( "warn: not english language: %s\n",
                                $attr->{lang} );
                            $self->eof();
                        }
                        elsif ( $tag eq 'meta'
                            and $attr->{name}
                            and lc( $attr->{name} ) eq 'robots' )
                        {
                            if (    $attr->{content}
                                and $attr->{content} =~ /nofollow/i )
                            {
                                print $log_fh sprintf( "warn: robots %s\n",
                                    $attr->{content} );
                                $no_follow_flag = 1;
                            }
                        }
                        elsif ( $tag eq 'frameset' ) {
                            print $log_fh "warn: frameset page\n";
                            $self->eof();
                        }
                    },
                    "self, tagname, attr"
                );

                last unless $html_parser->parse($content);

                $html_parser->eof();

                my $text = join '', @{ $html_parser->handler('text') };
                unless ( $text and $text !~ /^\s*$/ ) {
                    print $log_fh "warn: empty text\n";
                    last;
                }

                my $h_value = &get_SVM_value($text);

                unless ( defined $h_value ) {
                    print $log_fh "warn: empty significant text\n";
                    last;
                }

                my $digest =
                    Digest::MD5::md5_hex( Encode::encode_utf8($content) );
                $digests{$url} = $digest;
                if (@redirects) {
                    $digests{$_} = $digest for @redirects;
                }
                if ( exists $dumps{$digest} ) {
                    print $log_fh "warn: duplicated content!\n";
                    last;
                }
                my $label =
                    (defined $from ? ( $h_value > 0 ? '+' : '-' ) : '*');
                $download_cnt++;
                push @stat, $label;
                unless ( $label eq '*' ) {
                    $label eq '+' ? $target_cnt++ : $failure_cnt++;
                    print $log_fh "from: $from\n";
                    print $log_fh "label: $label\n";
                }

                if ( $package->can('post_fetch') ) {
                    $package->post_fetch( $url, $info, $label );
                }

                $dumps{$digest} = {
                    label     => $label,
                    index     => $download_cnt,
                    priority  => $priority,
                    relevance => $h_value,
                    base      => $base->as_string,
                    no_follow => $no_follow_flag,
                    content   => $content,
                };

                my $links_count = 0;
                unless ($no_follow_flag) {
                    $links_count =
                        $package->parse_postback( $content, $url, $base,
                        $label, $h_value, $path, $digest );
                    print $log_fh "links: $links_count\n";

                }
                else {
                    print $log_fh "links: meta tag nofollow\n";
                }
                unless ($params{silent}) {
                    print STDERR "[No.$download_cnt] $url ($label) [$elapsed_time Sec.] <$path> ", $links_count ? "(+: $links_count) " : '';
                    $package->display_additional_info($info) if $package->can('display_additional_info');
                    print STDERR "\n";
                }
            }
        }
        elsif ( $response->is_redirect ) {
            {
                my $location = $response->header('Location');
                print $log_fh "redirect to: $location\n" if $location;
                my $new_url = &get_canonical_url( $location, $base );
                last unless defined $new_url;
                my $new_host = &get_host($new_url);
                last if exists $urls{$new_url};
                unless ( @redirects < $params{max_redirect} ) {
                    print $log_fh "warn: too much redirects\n";
                    last;
                }
                push @$history, $url;
                $package->add_URL_to_frontier( $new_url, $REDIRECT,
                    [ $history, $path . 'R', @others ] );
            }
        }
    }
    print $log_fh sprintf( "time: %d Sec.\n", time - $now ) if time > $now;
    delete $busy_hosts{$host};
    $busy_hosts_cnt--;
    $_[KERNEL]->yield("ready");
}

sub display_state {
    my $package = shift;
    my $fh = $_[0] || *STDERR;
    print $fh '*' x 50, "\n";
    print $fh "Total     : $download_cnt\n";
    print $fh "Success   : $target_cnt\n";
    print $fh "Failure   : $failure_cnt\n";
    print $fh "Frontier  : $frontier_cnt\n";
    my $now   = time;
    my $count = 0;
    while ( my ( $host, $job ) = each %busy_hosts ) {
        $count++;
        print $fh sprintf "[No.%d]: %s (%d)\n", $count, $job->[0],
            $now - $job->[1];
    }
    if ( $target_cnt or $failure_cnt ) {
        print $fh sprintf "Total  Harvest Ratio    : (%.2f%%)\n",
            $target_cnt * 100 / ( $target_cnt + $failure_cnt );
        print $fh sprintf
            "Latest Harvest Ratio    : last 100 %s; last 500 %s; last 1000 %s.\n",
            &get_harvest_ratio(100), &get_harvest_ratio(500),
            &get_harvest_ratio(1000);
    }
    my $cursor = $frontier_db->db_cursor();
    my %hosts_seen;
    my ( $priority, $url ) = ( '', '' );
    while ( $cursor->c_get( $priority, $url, DB_NEXT ) == 0 ) {
        $hosts_seen{ &get_host($url) } = 1;
    }
    undef $cursor;
    print $fh sprintf "Hosts in frontier: %d\n", scalar keys %hosts_seen;
    print $fh '*' x 50, "\n";
}

sub get_canonical_url {
    my ( $href, $base ) = @_;
    return undef unless $href;
    my $uri = $base ? URI->new_abs( $href, $base ) : URI->new($href);
    $uri = $uri->canonical;
    my $canonical_url = lc $uri->as_string;
    return undef unless &is_HTML($canonical_url);
    return undef unless $uri->can('host') and defined $uri->host;
    return $canonical_url;
}

sub get_host {
    return URI->new(shift)->host;
}

sub is_HTML {
    my $url = shift;
    return 0 unless defined $url;
    return 0 unless $url =~ /^https?/;
    return 0 if $url =~ /[?#]/;
    my $depth = $url =~ tr/\//\//;
    return 0 if $depth > $params{max_url_depth};
    my ($suffix) = $url =~ /\.([^.]+)$/;
    if ($suffix) {
        return not exists $suffixes{$suffix};
    }
}

sub get_harvest_ratio {
    my $count = shift;
    my ( $pos, $neg ) = ( 0, 0 );
    $count = $count > @stat ? scalar @stat : $count;
    return 1 unless $count;
    for ( 1 .. $count ) {
        $pos++ if $stat[ -$_ ] eq '+';
        $neg++ if $stat[ -$_ ] eq '-';
    }
    return sprintf "%.2f%%(%d:%d)", $pos * 100 / ( $pos + $neg ), $pos, $neg;
}

sub get_SVM_value {
    my $tf_ref = TFIDF::text2tf(shift, \%classifier);
    my ($norm2, $sum) = (0, 0);
    while ( my ( $token, $tf ) = each %$tf_ref ) {
        my $pairRef = $classifier{$token};
        my $weight = $tf * $pairRef->[1];
        $norm2 += $weight**2;
        $sum   += $pairRef->[0] * $weight;
    }
    $norm2 = sqrt $norm2;
    return undef unless $norm2;
    return sprintf( "%.4f", $sum / $norm2 - $svm_bias );
}

sub add_URL_to_frontier {
    my $package = shift;
    my ( $url, $priority, $info ) = @_;
    $frontier_db->db_put( sprintf("%.4f", $priority), $url );
    $frontier_cnt++;
    $urls{$url} = $info;
}

sub get_new_url {
    my ( $anchor, $base ) = @_;
    my ( $href, $lang, $type ) = (
        $anchor->attr('href'), $anchor->attr('hreflang'),
        $anchor->attr('type')
    );
    return undef
        unless $href
        and $href !~ /\s+/
        and ( not defined $lang or $lang =~ /^en/i )
        and ( not defined $type or $type =~ m{^text/html}i );
    my $new_url = &get_canonical_url( $href, $base );
    return undef unless $new_url and not exists $urls{$new_url};
    return $new_url;
}

sub is_target_anchor {
    my ( $anchor, $base ) = @_;
    my $href = $anchor->attr('href');
    my $url = &get_canonical_url( $href, $base );
    return 0 unless defined $url;
    my $digest = $digests{$url};
    return 0 unless defined $digest;
    return $dumps{$digest}{label} ne '-';
}

sub init_frontier {
    my $package = shift;
    open SEEDS, "< $params{seeds_file}" or die "$!";
    while (<SEEDS>) {
        chomp;
        my $url = &get_canonical_url($_);
        next unless defined $url and not exists $urls{$url};
        $package->add_URL_to_frontier( $url, $SEED, [ [undef], 'S' ] );
    }
    close SEEDS;
    print STDERR sprintf "\nSeed URLs: %d\n", scalar keys %frontier unless $params{silent};
}

sub do_leaf_numbering_recursive {
    my ($elem, $ref) = @_;
    my @children = $elem->content_list;
    if (not @children) {
        $elem->{'sc_index'} = $$ref;
        $$ref++;
    }
    else {
        for my $child (@children) {
            next if $child->tag eq 'style' or $child->tag eq 'script';
            do_leaf_numbering_recursive($child, $ref);
        }
    }
}

sub collect_features {
    my ($anchor, $root, $result_ref, $last_index, $window_width) = @_;
    my ($min_leaf_no, $max_leaf_no) = ($last_index + 1, -1);
    sub_tree_min_max_recursive($anchor, \$min_leaf_no, \$max_leaf_no);
    collect_features_recursive($root, $min_leaf_no, $max_leaf_no, $window_width, $result_ref);
}

sub collect_features_recursive {
    my ($elem, $min_leaf_no, $max_leaf_no, $window_width, $result_ref) = @_;
    if (exists $elem->{'sc_index'}) {
        my $index = $elem->{'sc_index'};
        if ( $index >= $min_leaf_no - $window_width and $index <= $max_leaf_no + $window_width) {
            my $offset = 0;
            if ($index >= $min_leaf_no and $index <= $max_leaf_no) {
                $offset = 0;
            }
            if ($index < $min_leaf_no) {
                $offset = $index - $min_leaf_no;
            }
            if ($index > $max_leaf_no) {
                $offset = $index - $max_leaf_no;
            }
            push @$result_ref, [$offset, $elem->attr('text')];
        }
    }
    for my $child ($elem->content_list) {
        next if $child->tag eq 'style' or $child->tag eq 'script';
        collect_features_recursive($child, $min_leaf_no, $max_leaf_no, $window_width, $result_ref);
    }
}

sub sub_tree_min_max_recursive {
    my ($elem, $min_index_ref, $max_index_ref) = @_;
    my $index = $elem->{'sc_index'};
    if (defined $index) {
        if ($$max_index_ref < $index) {
            $$max_index_ref = $index;
        }
        if ($$min_index_ref > $index) {
            $$min_index_ref = $index;
        }
    }
    for my $child ($elem->content_list) {
        next if $child->tag eq 'style' or $child->tag eq 'script';
        sub_tree_min_max_recursive($child, $min_index_ref, $max_index_ref);
    }
}

sub check_db {
    my ($k, $v) = ("", "") ;
    my $cursor = $frontier_db->db_cursor() ;
    while ($cursor->c_get($k, $v, DB_NEXT) == 0)
    { print "$k -> $v\n" }
    undef $cursor ;
}

1;
